package org.lemon.transport.mqtt.server.session;

import com.google.protobuf.Descriptors;
import com.google.protobuf.DynamicMessage;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.util.ReferenceCountUtil;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.lemon.transport.context.data.device.DeviceProfile;
import org.lemon.transport.context.data.device.DeviceTransportType;
import org.lemon.transport.context.data.device.profile.DeviceProfileTransportConfiguration;
import org.lemon.transport.context.data.device.profile.MqttDeviceProfileTransportConfiguration;
import org.lemon.transport.context.data.device.profile.ProtoTransportPayloadConfiguration;
import org.lemon.transport.context.data.device.profile.TransportPayloadTypeConfiguration;
import org.lemon.transport.mqtt.server.model.MqttTopicMatcher;
import org.lemon.transport.mqtt.server.model.TopicType;
import org.server.transport.gen.SessionProto;
import org.lemon.transport.context.data.TransportPayloadType;
import org.lemon.transport.mqtt.server.adaptors.MqttTransportAdaptor;
import org.lemon.transport.mqtt.server.adaptors.MqttTransportAdaptorFactory;
import org.lemon.transport.mqtt.server.service.MqttTransportContext;
import org.lemon.transport.mqtt.util.MqttTopicFilterFactory;
import org.lemon.transport.mqtt.server.filter.MqttTopicFilter;

import java.util.Collection;
import java.util.Collections;
import java.util.UUID;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer;

@Slf4j
public class DeviceSessionContext extends MqttDeviceAwareSessionContext {

    @Getter
    @Setter
    private SessionProto.SessionInfoProto sessionInfo;
    @Getter
    @Setter
    private String clientId;
    @Getter
    @Setter
    private ChannelHandlerContext channel;
    @Getter
    private final MqttTransportContext context;

    private final ConcurrentLinkedQueue<MqttMessage> msgQueue = new ConcurrentLinkedQueue<>();
    private final AtomicInteger msgQueueSize = new AtomicInteger(0);
    private final Lock msgQueueProcessorLock = new ReentrantLock();

    /**
     * 设备是否只能发布消息
     */
    @Getter
    @Setter
    private boolean provisionOnly = false;

    private final AtomicInteger msgIdSeq = new AtomicInteger(0);

    private volatile MqttTopicFilter telemetryTopicFilter = MqttTopicFilterFactory.getDefaultTelemetryFilter();
    private volatile MqttTopicFilter attributesTopicFilter = MqttTopicFilterFactory.getDefaultAttributesFilter();


    private volatile boolean sendAckOnValidationException;
    private volatile boolean useJsonPayloadFormatForDefaultDownLinkTopics;

    private volatile MqttTransportAdaptor adaptor;

    @Getter
    private volatile TransportPayloadType payloadType = TransportPayloadType.JSON;
    private volatile Descriptors.Descriptor attributesDynamicMessageDescriptor;
    private volatile Descriptors.Descriptor telemetryDynamicMessageDescriptor;
    private volatile Descriptors.Descriptor rpcResponseDynamicMessageDescriptor;
    private volatile DynamicMessage.Builder rpcRequestDynamicMessageBuilder;

    private volatile boolean jsonPayloadFormatCompatibilityEnabled;
    private volatile boolean useJsonPayloadFormatForDefaultDownlinkTopics;
    @Getter
    @Setter
    private TransportPayloadType provisionPayloadType = payloadType;

    public DeviceSessionContext(UUID sessionId, ConcurrentMap<MqttTopicMatcher, Integer> mqttQoSMap, MqttTransportContext context) {
        super(sessionId, mqttQoSMap);
        this.adaptor = MqttTransportAdaptorFactory.getJsonMqttAdaptor();
        this.context = context;
    }

    public void addMessageToQueue(MqttMessage msg) {
        msgQueueSize.incrementAndGet();
        ReferenceCountUtil.retain(msg);
        msgQueue.add(msg);
    }

    public void tryProcessQueueMessage(Consumer<MqttMessage> msgProcessor) {

        while (!msgQueue.isEmpty()) {
            if (msgQueueProcessorLock.tryLock()) {
                try {
                    MqttMessage msg;
                    while ((msg = msgQueue.poll()) != null) {
                        try {
                            msgProcessor.accept(msg);
                            msgQueueSize.decrementAndGet();
                        } finally {
                            ReferenceCountUtil.safeRelease(msg);
                        }
                    }
                } finally {
                    msgQueueProcessorLock.unlock();
                }
            } else {
                return;
            }
        }
    }

    public boolean isDeviceTelemetryTopic(String topicName) {

        return telemetryTopicFilter.filter(topicName);
    }

    public boolean isDeviceAttributesTopic(String topicName) {
        return attributesTopicFilter.filter(topicName);
    }

    public boolean isSendAckOnValidationException() {
        return sendAckOnValidationException;
    }

    @Override
    public int nextMsgId() {
        return msgIdSeq.incrementAndGet();
    }

    public int getMsgQueueSize() {
        return msgQueueSize.get();
    }

    private void updateAdaptor() {
        if (payloadType.equals(TransportPayloadType.JSON)) {
            adaptor = MqttTransportAdaptorFactory.getJsonMqttAdaptor();
        } else if (payloadType.equals(TransportPayloadType.PROTOBUF)) {
            adaptor = MqttTransportAdaptorFactory.getProtoMqttAdaptor();
        }
    }

    public MqttTransportAdaptor getPayloadAdaptor() {
        return adaptor;
    }

    private void updateDeviceSessionConfiguration(DeviceProfile deviceProfile) {

        DeviceProfileTransportConfiguration transportConfiguration = deviceProfile.getProfileData().getTransportConfiguration();
        if (transportConfiguration.getType().equals(DeviceTransportType.MQTT)
                && transportConfiguration instanceof MqttDeviceProfileTransportConfiguration) {
            MqttDeviceProfileTransportConfiguration mqttConfig = (MqttDeviceProfileTransportConfiguration) transportConfiguration;
            TransportPayloadTypeConfiguration transportPayloadTypeConfiguration = mqttConfig.getTransportPayloadTypeConfiguration();
            payloadType = transportPayloadTypeConfiguration.getTransportPayloadType();
            telemetryTopicFilter = MqttTopicFilterFactory.toFilter(mqttConfig.getDeviceTelemetryTopic());
            attributesTopicFilter = MqttTopicFilterFactory.toFilter(mqttConfig.getDeviceAttributesTopic());
            sendAckOnValidationException = mqttConfig.isSendAckOnValidationException();
            if (TransportPayloadType.PROTOBUF.equals(payloadType)) {
                ProtoTransportPayloadConfiguration protoTransportPayloadConfig = (ProtoTransportPayloadConfiguration) transportPayloadTypeConfiguration;
                updateDynamicMessageDescriptors(protoTransportPayloadConfig);
                jsonPayloadFormatCompatibilityEnabled = protoTransportPayloadConfig.isEnableCompatibilityWithJsonPayloadFormat();
                useJsonPayloadFormatForDefaultDownlinkTopics = jsonPayloadFormatCompatibilityEnabled && protoTransportPayloadConfig.isUseJsonPayloadFormatForDefaultDownlinkTopics();
            } else {
                telemetryTopicFilter = MqttTopicFilterFactory.getDefaultTelemetryFilter();
                attributesTopicFilter = MqttTopicFilterFactory.getDefaultAttributesFilter();
                payloadType = TransportPayloadType.JSON;
                sendAckOnValidationException = false;
            }
            updateAdaptor();
        }
    }

    private void updateDynamicMessageDescriptors(ProtoTransportPayloadConfiguration protoTransportPayloadConfig) {
        telemetryDynamicMessageDescriptor = protoTransportPayloadConfig.getTelemetryDynamicMessageDescriptor(protoTransportPayloadConfig.getDeviceTelemetryProtoSchema());
        attributesDynamicMessageDescriptor = protoTransportPayloadConfig.getAttributesDynamicMessageDescriptor(protoTransportPayloadConfig.getDeviceAttributesProtoSchema());
        rpcResponseDynamicMessageDescriptor = protoTransportPayloadConfig.getRpcResponseDynamicMessageDescriptor(protoTransportPayloadConfig.getDeviceRpcResponseProtoSchema());
        rpcRequestDynamicMessageBuilder = protoTransportPayloadConfig.getRpcRequestDynamicMessageBuilder(protoTransportPayloadConfig.getDeviceRpcRequestProtoSchema());
    }

    public MqttTransportAdaptor getAdaptor(TopicType topicType) {
        switch (topicType) {
            case V2:
                return getDefaultAdaptor();
            case V2_JSON:
                return MqttTransportAdaptorFactory.getJsonMqttAdaptor();
            case V2_PROTO:
                return MqttTransportAdaptorFactory.getProtoMqttAdaptor();
            default:
                return useJsonPayloadFormatForDefaultDownLinkTopics ?
                        MqttTransportAdaptorFactory.getJsonMqttAdaptor() : getDefaultAdaptor();
        }
    }

    public void release() {
        if (!msgQueue.isEmpty()) {
            log.warn("doDisconnect for device {} but unprocessed messages {} left in the msg queue", getDeviceId(), msgQueue.size());
            msgQueue.forEach(ReferenceCountUtil::safeRelease);
            msgQueue.clear();
        }
    }

    public Collection<MqttMessage> getMsgQueueSnapshot() {
        return Collections.unmodifiableCollection(msgQueue);
    }

    private MqttTransportAdaptor getDefaultAdaptor() {
        return payloadType.equals(TransportPayloadType.JSON) ?
                MqttTransportAdaptorFactory.getJsonMqttAdaptor() : MqttTransportAdaptorFactory.getProtoMqttAdaptor();
    }

    public Descriptors.Descriptor getTelemetryDynamicMsgDescriptor() {
        return telemetryDynamicMessageDescriptor;
    }

    public Descriptors.Descriptor getAttributesDynamicMessageDescriptor() {
        return attributesDynamicMessageDescriptor;
    }

    public Descriptors.Descriptor getRpcResponseDynamicMessageDescriptor() {
        return rpcResponseDynamicMessageDescriptor;
    }

    public DynamicMessage.Builder getRpcRequestDynamicMessageBuilder() {
        return rpcRequestDynamicMessageBuilder;
    }
}
